package net.xuele.learn.flink.example;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


import java.sql.PreparedStatement;

public class JdbcSinkExample {
    public static void main(String[] args) throws Exception {
        // 设置批处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();



        DataStream<Tuple3<Integer, String, Integer>> dataStream = env.fromElements(new Tuple3<>(1, "Alice", 25),
                new Tuple3<>(2, "Bob", 30),
                new Tuple3<>(3, "Charlie", 22));

        // 将 Tuple3 转换为 Person 类型（你的数据类）
        DataStream<Person> personStream = dataStream.map(new MapFunction<Tuple3<Integer, String, Integer>, Person>() {
            @Override
            public Person map(Tuple3<Integer, String, Integer> value) throws Exception {
                return new Person(value.f0, value.f1, value.f2);
            }
        });

        // JdbcSink 配置
        String dbUrl = "jdbc:mysql://localhost:3306/mydatabase";
        String username = "your_username";
        String password = "your_password";

        personStream.addSink(JdbcSink.sink(
                "INSERT INTO people (id, name, age) VALUES (?, ?, ?)",   // SQL 插入语句
                (PreparedStatement statement, Person person) -> {      // PreparedStatementSetter
                    statement.setInt(1, person.getId());
                    statement.setString(2, person.getName());
                    statement.setInt(3, person.getAge());
                },
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()  // 数据库连接配置
                        .withDriverName("com.mysql.jdbc.Driver")
                        .withUrl(dbUrl)
                        .withUsername(username)
                        .withPassword(password)
                        .build()
        ));

        // 执行程序
        env.execute("JdbcSink Example");
    }
}
